package com.oracle.violet.first.area

import com.oracle.violet.first.config.ConfigHandler
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object AreaDist {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("地域分布")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)

    //读取文件
    val dataFrame: DataFrame = sQLContext.read.parquet(ConfigHandler.parquetFilePath)
    dataFrame.registerTempTable("t_person")
    //原始请求数
    sQLContext.sql(
      """
        | select provincename, cityname,
        |sum(if(requestmode=1 and processnode>=1, 1, 0)) adRawReq,
        |sum(if(requestmode=1 and processnode>=2, 1, 0)) adEffReq,
        |sum(case when requestmode=1 and processnode=3 then 1 else 0 end) adReq,
        |sum(case when iseffective=1 and isbilling=1 and isbid=1 and adorderid!=0 then 1 else 0 end) adRtbReq,
        |sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) adSuccReq,
        |sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) adShow,
        |sum(case when requestmode=3 and iseffective=1 then 1 else 0 end) adClick,
        |sum(case when iseffective=1 and isbilling=1 and iswin=1 then winprice/1000 else 0 end) adCost,
        |sum(case when iseffective=1 and isbilling=1 and iswin=1 then adpayment/1000 else 0 end) adConsumption
        |from t_person
        |group by provincename,cityname

      """.stripMargin)
      .write
      .jdbc(ConfigHandler.url, "orc_AreaDist_province", ConfigHandler.props)


    sc.stop()
  }

}
